package Farsail::Bulkload;

use strict; 
use warnings;

use Carp;
use File::Glob qw/:glob/;
use File::Temp qw/tempfile/;
use Farsail;
use Path::Class;
use Farsail::Util qw/expand_file/;

my ($farsail, $logger);
my $PGQUOTE = '$$';
my @FIELDS = qw( TABLE INFILE TYPE MAX_ERR_CNT OFFSET LIMIT );

sub init {
    my $pkg = shift;
    $farsail = shift;
    $logger = Log::Log4perl->get_logger();
}

sub log_do {
    my ($dbh, $sql) = @_;
    $logger->debug($sql);
    $dbh->do($sql) or $logger->logdie($sql);
}

sub read_config {
    my $file = shift;
    open(my $fh, "<", $file)
        or $logger->logdie("Can't open file $file: $!");
    my %config;
    while ( <$fh> ) {
		next if /^\s*#/;
		next if /^\s*$/;
		chomp;
        s/#.*$//;
        s/\s+$//;
        s/^\s+//;
		my ($key,$value) = split /\s*=\s*/;
		$config{$key} = $value;
    }
    return \%config;
}

sub write_config {
    my ($opt, $dir) = @_;
    for ( qw/file table format/ ) {
        if ( !exists $opt->{$_} ) {
            confess "No option $_\n";
        }
    }
    my %config = (
        TABLE => $opt->{table},
        INFILE => $opt->{file},
    );
    my $format;
    if ( ref $opt->{format} eq 'HASH' ) {
        $format = $opt->{format};
    } else {
        my $config = $farsail->config;
        my $tplfile = expand_file(
            $opt->{format},
            $config->get_expand_file('bulkload.format_directory'),
        );
        $format = read_config($tplfile);
    }
    for ( keys %$format ) {
        next if exists $config{$_};
        $config{$_} = $format->{$_}; 
    }
    my ($fh, $fn) = tempfile( DIR => $dir );
    for ( @FIELDS ) {
        if ( exists $config{$_} ) {
            print {$fh} "$_ = ", delete $config{$_}, "\n";
        }
    }
    for ( keys %config ) {
        print {$fh} "$_ = $config{$_}\n";
    }
    close($fh);
    chmod(0644, $fn) or die "chmod file '$fn' failed: $!";
    return $fn;
}

sub expand_template {
    my ($tmpl, $vars) = @_;
    for ( sort {$b cmp $a} keys %$vars ) {
        $tmpl =~ s/\$$_/$vars->{$_}/msge;
    }
    return $tmpl;
}

sub expand_sql {
    my $opt = shift;
    return expand_template($opt->{create_sql}, {table => $opt->{table}});
}

sub ACTION_bulkload {
    my $self = shift;
    my %args = $farsail->args->varlist('.*');
    $self->bulkload(%args);
    return 1;
}

sub bulkload {
    my $self = shift;
    my %opt = @_;
    if ( !exists $opt{format} ) {
        confess "No option format!\n";
    }
    if ( $opt{file}  ) {
        bulkload_file(\%opt);
    } else {
        bulkload_files(\%opt)
    }
}

sub bulkload_files {
    my $opt = shift;
    for ( qw/dir cluster table/ ) {
        confess "No option $_!\n" if !$opt->{$_};
    }
    my $config = $farsail->config;
    $config->set('bulkload.done_file', '');
    Farsail::PBS::Job->set_default_options(
        -N => 'bulkload',
        -W => 'umask=000',
    );
    my $pbs = $farsail->pbs();
    my $dir = $opt->{dir};
    my $prefix = (defined $opt->{prefix} ? $opt->{prefix} : 'bucket');
    my $suffix = (defined $opt->{suffix} ? $opt->{suffix} : 'dat');
    my $glob_pattern = "$dir/${prefix}[0-9]*" . ( $suffix ? ".$suffix" : "");
    my $tmpdir = $config->get_expand_file('bulkload.share_directory');
    my @files = bsd_glob($glob_pattern, GLOB_ERROR);
    if ( @files ) {
        my $cluster = $config->cluster($opt->{cluster});
        my $nodes = $cluster->{nodes};
        my $has_local_bucket = 0;
        if ( @files > @$nodes ) {
            $has_local_bucket = 1;
        }
        my $table = $opt->{table};
        my @jobs;
        for ( @files ) {
            $opt->{file} = $_;
            my $name = file($_)->basename;
            $name =~ /\Q$prefix\E(\d+)/;
            my $num = $1;
            if ( $has_local_bucket ) {
                $opt->{table} = sprintf("%s_%d", $table, int($num/scalar(@$nodes))+1);
            }
            my $ctlfile = write_config( $opt, $tmpdir );
            my $cmd = "";
            my $programs = $config->section('programs');
            my $db = $config->database($nodes->[$num%scalar(@$nodes)]);
            my %opts = $db->psql_options();
            my $host = delete $opts{-h};
            my $psql = $programs->{psql} || "psql";
            if ( $opt->{truncate} ) {
                $cmd = join(" ", $psql, %opts, "-c", "'TRUNCATE TABLE $opt->{table}'") . "\n";
            } elsif ( $opt->{create_sql} ) {
                $cmd = join(" ", $psql, %opts) . "<<SQL\n" . expand_sql($opt) . "\nSQL\n";
            }
            my $pg_bulkload = $programs->{pg_bulkload} || "pg_bulkload";
            $cmd .= join(" ", $pg_bulkload, %opts, $ctlfile);
            my $job = $pbs->add_job($cmd);
            $job->set_options(-l => "nodes=$host");
            push @jobs, $job;
        }
        my $done = file($tmpdir, "$$.done");
        $done->remove();
        $pbs->add_job("touch $done", \@jobs);
        $pbs->submit();
        $config->set('bulkload.done_file', $done);
    } else {
        confess "No files found with pattern: $glob_pattern" unless $farsail->args->fake;
    }
}

sub bulkload_file {
    my $opt = shift;
    for ( qw/file table /) {
        confess "No option $_!\n" if !$opt->{$_};
    }
    my ($db, $cluster);
    if ( $opt->{database} ) {
        $db = $opt->{database};
    } elsif ( $opt->{cluster} ) {
        $cluster = $opt->{cluster};
    } else {
        confess "No option database or cluster\n";
    }
    if ( !-e $opt->{file} ) {
        confess "File '$opt->{file}' does not exists!\n" unless $farsail->args->fake;
    }
    my $config = $farsail->config;
    my $dbh = ( $db ? $farsail->db($db) : $farsail->cluster($cluster));
    if ( $opt->{truncate} ) {
        log_do($dbh, ($db ? "TRUNCATE TABLE $opt->{table}"
                          : "SELECT ddlexec($PGQUOTE TRUNCATE TABLE $opt->{table} $PGQUOTE)"));
    }
    elsif ( $opt->{create_sql} ) {
        my $sql = expand_sql($opt);
        log_do($dbh, ($db ? $sql : "SELECT ddlexec($PGQUOTE $sql $PGQUOTE)"));
    }
    my $ctlfile = write_config( $opt, $config->get_expand_file('bulkload.share_directory'));
    log_do($dbh, ($db ? "SELECT pg_bulkload('$ctlfile')"
              : "SELECT ddlexec($PGQUOTE SELECT pg_bulkload('$ctlfile') $PGQUOTE)"));
}

1;
